package org.ilc.sample;

import com.mongodb.*;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.ilc.Bootstrap;
import org.ilc.push.IDataProvider;
import org.ilc.push.Message;
import org.ilc.sample.cache.CacheExecutor;
import org.ilc.sample.cache.CacheUtil;
import org.ilc.sample.util.BeanUtil;
import org.ilc.sample.util.DBUtil;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;

/**
 * Data cache org.ilc.push.impl
 * @author malb
 */
@SuppressWarnings("ALL")
public class SimpleDataProvider implements IDataProvider {
    private static Log log = LogFactory.getLog ("SimpleDataProvider");

    private static final long MESSAGE_EXP = 0;
    private static String MESSAGE_QUEUE_Key = "msgqueue";
    private DB db = null;

    private CacheExecutor executor = null;

    private static int expiredTime = 1800000;
    private static String dbName = "ilcdb";

    static {
        expiredTime = Bootstrap.getContext ().getAppConfig ().getInt (Constants.MESSAGE_EXPAIRE_TIME_KEY, 1800000);
        dbName = Bootstrap.getContext ().getAppConfig ().get (Constants.DB_NAME_KEY);
        if (dbName == null || "".equals (dbName)) {
            log.warn ("参数：" + Constants.DB_NAME_KEY + "无效，将使用默认值：ilcdb");
            dbName = "ilcdb";
        }
    }
    /**
     *
     */
    public SimpleDataProvider() {
        db = DBUtil.getDB (dbName);
        executor = new CacheExecutor ();

    }

    /**
     *
     * @return
     */
    @Override
    public Message next() {
        Message msg = null;
        String createTimeStr = "";
        try {
            String cacheValue = CacheUtil.get ();
            if (cacheValue == null)
                return null;

            String [] values = cacheValue.split (":");
            if (values == null || values.length <= 0)
                return null;

            int firstIndex = cacheValue.indexOf (":");
            if (firstIndex < 0)
                return null;

            int secondIndex = cacheValue.indexOf (":", firstIndex+1);

            if (secondIndex < 0)
                return null;

            String user = cacheValue.substring (0, firstIndex);
            createTimeStr = cacheValue.substring (firstIndex+1, secondIndex);
            String content = cacheValue.substring (secondIndex+1);

            if (user == null || "".equals (user) || content == null) return null;
            msg = newMessageEntity (user , content);
        } catch (Exception e) {
            if (log.isErrorEnabled()) {
                log.error("get message from queue failed.",e);
            }
            return null;
        }

        try {
            long createTime = Long.parseLong(createTimeStr);
            long nowTime = System.currentTimeMillis();
            //检查消息是否过期
            if (nowTime - createTime > expiredTime)
                return null;

            msg.setExpiredTime (expiredTime);
            msg.setCreateTime (createTime);
        } catch (Exception e) {
            //ignore
            return null;
        }
        return msg;
    }

    /**
     *
     * @param user
     * @param content
     */
    @Override
    public void cache(String user, String content) {
        try {
            long createTime = System.currentTimeMillis ();
            cache (user, content, createTime);
        } catch (Exception e) {
            if (log.isErrorEnabled()) {
                log.error("cache message failed.", e);
            }
        }
    }

    public void cache (String user , String content, long createTime) {
        if (user == null || "".equals (user) || content == null) return;
        String value = "";
        value = value.concat (user).concat (":").
                concat (createTime + "").concat (":").
                concat (content);
//        System.out.println (value);

        executor.cache (value);
    }


    /**
     *
     * @param msg
     */
    @Override
    public void store(Message msg) {
        if (msg != null) {
            System.out.println ("send failed:" + msg.getAccount () + "," + msg.getContent ());
            DBObject dbObject = null;
            DBCollection messageCollection = null;
            try {
                messageCollection = db.getCollection("messages");
                //Convert message to DBObject
                msg.setStatus (Message.STATUS_SEND_FAILUER);
                dbObject = BeanUtil.bean2DBObject (msg);


                //cache message to mongodb
                if (dbObject != null && messageCollection != null) {
                        messageCollection.insert(dbObject);
                }
            } catch (Exception e) {
                if (log.isErrorEnabled())
                    log.error("Store message failed", e);
                return;
            }
        }
    }

    /**
     *
     * @param user
     */
    @Override
    public void cacheFromStore(String user) {
        //cache message where get from mongodb
        List<Message> messages = new ArrayList<Message>();

        try {
            if (user != null && !"".equals(user)) {
                DBObject dbObject = null;
                Message msg = null;
                DBCollection messageCollection = db.getCollection("messages");
                //get data from mongodb by account
                DBCursor dbCursor = messageCollection.find(new BasicDBObject("account", user));

                /*
                Convert DBObjects to Messages
                 */
                while (dbCursor.hasNext()) {
                    dbObject = dbCursor.next();
                    try {
                        if (dbObject == null) continue;

    //                    System.out.print ("from mongod:");
                        dbCursor.count ();
                        dbObject.get ("account");
                        msg = BeanUtil.DBObject2Message (dbObject);
                        messages.add(msg);
                        messageCollection.remove (dbObject);
                    } catch (Exception e) {
                        if (log.isErrorEnabled())
                            log.error("Convert DBObject to Message failed", e);
                    }
                }
            }

            /*
            Cache message to Memcached.
             */
            for (Message msg : messages) {
                cache (msg.getAccount (),msg.getContent (), msg.getCreateTime ());
            }
        } catch (Exception e) {
            if (log.isErrorEnabled()) {
                log.error("cache from store failed.",e);
            }
        }
    }

    /**
     *
     * @param e
     */
    private void logError (Throwable e) {
        if (log.isErrorEnabled())
            log.error ((new Date()).toString (), e);
    }

    /**
     *
     * @param user
     * @param content
     * @return
     */
    private Message newMessageEntity(String user, String content) {
        Message msg = new Message ();
        msg.setAccount (user);
        msg.setContent (content);
        msg.setCreateTime (System.currentTimeMillis());
        msg.setStatus (Message.STATUS_WAIT);
        msg.setExpiredTime (-1);

        return msg;
    }

}
